package com.wyj;

import com.alibaba.fastjson.JSONObject;
import com.wyj.mapper.MessageMapper;
import com.wyj.pojo.Message;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.checkerframework.checker.units.qual.C;
import org.checkerframework.dataflow.qual.TerminatesExecution;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.HashMap;
import java.util.Map;

/**
 * @author 王永杰
 * @Description 描述该类的做什么用
 * @since 2022/8/18
 */
@RunWith(SpringRunner.class)
@SpringBootTest
@ContextConfiguration(classes = NiukeltApplication.class)
@Log4j2
public class textKafka {

    @Autowired
    KafkaProducer1 kafkaProducer;

    @Autowired
    MessageMapper messageMapper;


    @Test
    public void text01() throws InterruptedException {
        kafkaProducer.sendMessage("text", "hello");
        kafkaProducer.sendMessage("text", "my name is wyj");

        Thread.sleep(1000 * 10);
    }

    @Test
    public void text02() {
        System.out.println("hello");
    }

    @Test
    public void textMessageMapper1() {
        Message message = messageMapper.selectLatestNotice(111, "comment");

        int comment = messageMapper.selectCountsNotices(111, "comment");
        int comment1 = messageMapper.selectCountsNoReadNotices(111, "comment");

        System.out.println("总共的comment的信息" + comment);
        System.out.println("未读的comment1的信息"+ comment1);
    }


    @Test
    public void textMessage01() {

        Message message = messageMapper.selectLatestNotice(111, "comment");

//        Map<String,Object> map = JSONObject.parseObject(message.getContent(), HashMap.class);

//        for (String s : map.keySet()) {
//            System.out.println(s+":"+map.get(s));
//        }

    }


}

@Component
class KafkaProducer1 {
    @Autowired
    private KafkaTemplate kafkaTemplate;

    public void sendMessage(String topic, String content) {
        kafkaTemplate.send(topic, content);
    }
}

@Component
class kafkaConsumer {

    @KafkaListener(topics = {"text"})
    public void receiveMessage(ConsumerRecord record) {
        System.out.println(record.value());
    }
}
